package org.huel.backend.websocket;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.huel.backend.config.WebSocketServer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.*;

/**
 * @Kafka topic: LACountByClass
 * @author Liweijian.
 * @Description 每个班级的L&A
 * @date 2024/11/17 10:45
 */
@Component
public class LACountByClassConsumer {

    private static final Map<String, int[]> classCounts = new HashMap<>();
    private KafkaConsumer<String, String> kfkConsumer;

    @PostConstruct
    public void init() {
        // 消费者配置信息
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "likai:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "LACountByClassConsumer");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 构建 Kafka 消费者
        kfkConsumer = new KafkaConsumer<>(properties);
        kfkConsumer.subscribe(Collections.singletonList("LACountByClass1"));

        // 启动消费者线程
        new Thread(this::consumeMessages).start();
    }

    private void consumeMessages() {
        try {
            while (true) {
                // 通过 poll 的形式拉取
                ConsumerRecords<String, String> conRecord = kfkConsumer.poll(Duration.ofMillis(2000));

                // 清空 classCounts，为了不累加
                Map<String, int[]> currentCount = new HashMap<>();

                for (ConsumerRecord<String, String> consumer : conRecord) {
                    // 解析 JSON
                    JSONObject jsonObject = JSONObject.parseObject(consumer.value());
                    String classId = jsonObject.getString("classId");
                    String status = jsonObject.getString("status");
                    int count = jsonObject.getInteger("count"); // 获取当前记录的 count 值

                    // 更新计数
                    currentCount.putIfAbsent(classId, new int[2]); // [0] for 'L', [1] for 'A'
                    if ("L".equals(status)) {
                        currentCount.get(classId)[0] = count; // 将当前出勤数设置为 count
                    } else if ("A".equals(status)) {
                        currentCount.get(classId)[1] = count; // 将当前缺勤数设置为 count
                    }
                }

                // 生成输出 JSON
                JSONArray outputJason = new JSONArray();
                for (Map.Entry<String, int[]> entry : currentCount.entrySet()) {
                    String id = entry.getKey();
                    int countL = entry.getValue()[0];
                    int countA = entry.getValue()[1];

                    JSONObject classStatus = new JSONObject();
                    classStatus.put("classId", id);
                    classStatus.put("countL", countL);
                    classStatus.put("countA", countA);

                    outputJason.add(classStatus);
                }

//                // 输出当前状态
//                System.out.println("================");
//                System.out.println(outputJason.toJSONString());
//                System.out.println("================");

                WebSocketServer.sendMessage("LACountByClass", String.valueOf(outputJason));
            }
        } catch (Exception e) {
            e.printStackTrace(); // 处理异常并输出错误信息
        } finally {
            // 确保消费者在结束时关闭
            kfkConsumer.close();
        }
    }

}